package com.itttq.im.ws.netty;

import com.itttq.im.common.constant.Constant;
import com.itttq.im.common.constant.RedisConstant;
import com.itttq.im.conf.WsProperties;
import com.itttq.im.model.to.MsgTo;
import com.itttq.im.utils.GenUtil;
import com.itttq.im.utils.RedisUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 业务处理类：客户端的请求会在这里处理。比如客户端连接、客户端发送消息、给客户端发送消息
 * TextWebSocketFrame是netty用于处理websocket发来的文本对象
 * 如果需要在助手类中用到@Autowire注解，可以参考这个博客：
 * https://blog.csdn.net/weixin_30828379/article/details/95009595
 * <p>
 * 每个成功的线程访问顺序：channelActive（开启连接）-handleHttpRequest（http握手处理）-messageReceived（消息接收处理）-handlerWebSocketFrame（实际处理，可以放到其他类里面分业务进行）
 *
 * @Author GXG
 */
@ChannelHandler.Sharable
@Component
@Slf4j
public class WsServerHandler extends SimpleChannelInboundHandler<Object> {
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private MessageHandler messageHandler;

    /**
     * 所有正在连接的channel都会存在这里面，所以也可以间接代表在线的客户端
     */
    public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    /**
     * 在线人数
     */
    public static int online = 0;
    private WebSocketServerHandshaker handshaker;

    // TODO 下面的可以放ChannelHandle类
    /**
     * 存储频道和用户的关联关系：key为userId
     */
    public static ConcurrentHashMap<String, Channel> userChannel = new ConcurrentHashMap<>();
    public static WsServerHandler wsServerHandler;

    /**
     * 存用户和群组的关联关系：key为groupId，值为分组的用户ID集合
     */
    public static ConcurrentHashMap<String, ChannelGroup> groupChannels = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        wsServerHandler = this;
    }

    /**
     * 客户端与服务器建立连接的时候触发
     * 当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("与客户端建立连接，通道开启！");
        //添加到channelGroup通道组
        channelGroup.add(ctx.channel());
        online = channelGroup.size();
        log.info("{}上线，当前人数：{}", ctx.channel().remoteAddress(), online);
    }

    /**
     * 客户端与服务器关闭连接的时候触发
     * 当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端关闭了通信通道并且不可以传输数据
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("与客户端断开连接，通道关闭！");
        releaseChannel(ctx.channel());
    }

    /**
     * 在读操作异常被抛出时被调用
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("WebSocket出错");
        releaseChannel(ctx.channel());
        ctx.fireExceptionCaught(cause);
        ctx.close();
    }

    /**
     * 用来通知handler上一个ChannelRead()是被这批消息中的最后一个消息调用
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    /**
     * 服务器接受客户端的数据信息
     *
     * @param ctx
     * @param msg
     */
    @Override
    public void channelRead0(ChannelHandlerContext ctx, Object msg) {
        //  每次都重新设置空闲时间
        ctx.channel().attr(Attributes.IDLE_TIME).set(System.currentTimeMillis());
        if (null == msg) {
            return;
        }
        if (msg instanceof FullHttpRequest) {
            // 处理websocket握手
            FullHttpRequest request = (FullHttpRequest) msg;
            handleHttpRequest(ctx, request);
        } else if (msg instanceof WebSocketFrame) {
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof CloseWebSocketFrame) {
            // 关闭链路的指令
            releaseChannel(ctx.channel());
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
        } else if (frame instanceof PingWebSocketFrame) {
            // ping/pong作为心跳
            ctx.write(new PongWebSocketFrame(frame.content().retain()));
        } else if (frame instanceof TextWebSocketFrame) {
            // 正常的业务消息
            TextWebSocketFrame request = (TextWebSocketFrame) frame;
            // 接收到正常的消息
            wsServerHandler.messageHandler.handleMsg(ctx.channel(), request.text());
        } else if (frame instanceof BinaryWebSocketFrame) {
            // 不处理二进制消息
            ctx.write(frame.retain());
        }
    }

    /**
     * 关闭时，释放频道资源
     *
     * @param channel
     */
    private void releaseChannel(Channel channel) {
        String uid = channel.attr(Attributes.USER_ID).get();
        if (StringUtils.isNotEmpty(uid)) {
            // 移除用户组
            userChannel.remove(uid);
            // 清除缓存内容
            redisUtil.remove(RedisConstant.getKeyUserOnline(), uid);
            redisUtil.delete(RedisConstant.getKeyUserServer(), uid);
        }
        // 移除群聊组中自己的频道
        String groupId = channel.attr(Attributes.GROUP_ID).get();
        if (StringUtils.isNotEmpty(groupId)) {
            ChannelGroup groupChannel = groupChannels.get(groupId);
            if (!CollectionUtils.isEmpty(groupChannel)) {
                groupChannel.remove(channel);
            }
            redisUtil.remove(RedisConstant.getKeyGroupId(), groupId);
            redisUtil.remove(RedisConstant.getKeyGroupUsers(groupId), uid);
        }
        // 移除频道
        channelGroup.remove(channel);
        online = channelGroup.size();
        // 给所有人发，含自己
        messageHandler.sendGroupMsg(MsgTo.builder().body(uid + "下线啦").type(MsgTo.MsgType.BROADCAST.getCode()).build());
    }

    /**
     * 首次请求为HTTP，用于创建websocket
     *
     * @param ctx
     * @param request
     */
    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
        //要求Upgrade为websocket，过滤掉get/Post
        if (!request.decoderResult().isSuccess() || (!"websocket".equals(request.headers().get("Upgrade")))) {
            //若不是websocket方式，则创建BAD_REQUEST的req，返回给客户端
            sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        String uri = request.uri();
        log.info(uri);
        // 记录用户ID
        String uid = GenUtil.getParamFromUri(uri, "uid");
        if (StringUtils.isEmpty(uid)) {
            sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.UNAUTHORIZED));
        } else {
            ctx.channel().attr(Attributes.USER_ID).set(uid);
            userChannel.put(uid, ctx.channel());
            // 记录到Redis
            redisUtil.sSet(RedisConstant.getKeyUserOnline(), uid);
            redisUtil.put(RedisConstant.getKeyUserServer(), uid, WsProperties.localIp + Constant.COLON + WsProperties.httpPort);
            // 给所有人发，含自己
            messageHandler.sendGroupMsg(MsgTo.builder().body(uid + "上线啦").type(MsgTo.MsgType.BROADCAST.getCode()).build());
        }
        // 记录分组ID
        String groupId = GenUtil.getParamFromUri(uri, "gid");
        if (StringUtils.isNotEmpty(groupId)) {
            ChannelGroup groupChannel = groupChannels.get(groupId);
            if (null == groupChannel) {
                groupChannel = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
            }
            groupChannel.add(ctx.channel());
            groupChannels.put(groupId, groupChannel);
            ctx.channel().attr(Attributes.GROUP_ID).set(groupId);
            redisUtil.sSet(RedisConstant.getKeyGroupId(), groupId);
            redisUtil.sSet(RedisConstant.getKeyGroupUsers(groupId), uid);
        }
        log.info("群组用户数：{}", groupChannels.get(groupId).size());
        // 获取用户参数，作为唯一标识使用
        // TODO 此处设置
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://" + request.headers().get(org.apache.http.HttpHeaders.HOST) + uri, null, false);
        handshaker = wsFactory.newHandshaker(request);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), request);
        }
    }

    /**
     * 拒绝不合法的请求，并返回错误信息
     */
    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
        /**
         *  返回应答给客户端
         */
        if (HttpStatus.SC_OK != res.status().code()) {
            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
        }
        /**
         * 如果是非Keep-Alive，关闭连接
         */
        if (!HttpUtil.isKeepAlive(req) || HttpStatus.SC_OK != res.status().code()) {
            ChannelFuture f = ctx.channel().writeAndFlush(res);
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

}
